package source


import bean.SensorReading
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.functions.source.SourceFunction
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}

import scala.util.Random

/**
  * @Description: TODO QQ1667847363
  * @author: xiao kun tai
  * @date:2021/11/14 14:38
  */

object Source4_UDF {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val stream: DataStream[SensorReading] = env.addSource(new MySensorSource())
    stream.print()

    env.execute()


  }

  //自定义SourceFunction
  class MySensorSource() extends SourceFunction[SensorReading] {
    var running: Boolean = true

    override def run(sourceContext: SourceFunction.SourceContext[SensorReading]): Unit = {

      //定义一个随机数发生器(id,temp)
      val rand = new Random


      //随机生成一组（10）传感器的初始温度
      var curTemp = 1.to(10).map(i => ("sensor_" + i, rand.nextDouble() * 100))

      //定义无限循环产生数据，不停地产生数据，除非被cancel取消
      while (running) {
        //上次数据基础上微调，更新温度值
        curTemp = curTemp.map(
          data => (data._1, data._2 + rand.nextGaussian())
        )
        //获取当前时间戳，加入数据中
        val curTime = System.currentTimeMillis()

        curTemp.foreach(data =>
          sourceContext.collect(SensorReading(data._1, curTime, data._2)))
        //间隔100ms
        Thread.sleep(100)
      }

    }

    override def cancel(): Unit = running = false
  }

}




